package com.xctech.yace.swing.subTab;

import com.xctech.yace.service.ExcelService;
import com.xctech.yace.service.FileService;
import com.xctech.yace.service.KafkaRecived;
import com.xctech.yace.single.RecivedDataManger;
import com.xctech.yace.swing.ConfigValue;
import com.xctech.yace.swing.FatherSwing;
import com.xctech.yace.swing.InitFileConfig;
import com.xctech.yace.thread.ConsumerThread;
import com.xctech.yace.util.DirUtil;
import com.xctech.yace.util.FormatUtil;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;

import javax.swing.*;
import java.awt.*;
import java.awt.event.*;

import java.time.Instant;
import java.time.ZoneId;
import java.util.List;
import java.util.Properties;
import java.util.Queue;

import static javax.swing.JOptionPane.WARNING_MESSAGE;

@Slf4j
public class MonitorSwing extends FatherSwing implements ConfigValue, InitFileConfig {

    public static Boolean isMonitor = false;

    //界面刷新线程
    public static Thread flush = null;
    //监听线程
    public static Thread reciving = null;

    //接收类
    public static KafkaRecived recived = new KafkaRecived();

    public static void placeComponentsRight(JComponent panel) {
        /* 布局部分我们这边不多做介绍
         * 这边设置布局为 null
         */
        panel.setLayout(null);

        Properties properties = consumerPropties();
        String ipport = properties.getProperty("bootstrap.servers");

        JComboBox topicBox=new JComboBox();    //创建JComboBox
        topicBox.addItem("--自定义主题--");    //向下拉列表中添加一项
        topicMap.entrySet().stream().forEach(topics->{
            topicBox.addItem(topics.getKey());
        });
        topicBox.setToolTipText(ipport);
        topicBox.setBounds(15,20,sfiledW,height);
        panel.add(topicBox);

        JTextField topicText = new JTextField(20);
        topicText.setBounds(15+sfiledW+space-10,20,300,height);
        topicText.setFont(new Font("Courier",Font.BOLD,10));
        topicText.setToolTipText(ipport);
        panel.add(topicText);

        // 状态显示
        JTextField statusText = new JTextField(20);
        statusText.setBounds(550,20 , 115, height);
        statusText.setFont(new Font("Courier",Font.BOLD,13));
        statusText.setToolTipText(ipport);
        panel.add(statusText);

        // 监听按钮
        JButton startButton = new JButton("开启监听");
        startButton.setBounds(440,20,100,height);
        startButton.setToolTipText(ipport);
        panel.add(startButton);

        // 停止按钮
        JButton stopButton = new JButton("停止监听");
        stopButton.setBounds(440,20,100,height);
        stopButton.setVisible(false);
        stopButton.setToolTipText(ipport);
        panel.add(stopButton);


        //message按钮
        JTextArea area = new JTextArea();
        JScrollPane scrollArea = new JScrollPane(area);

        scrollArea.setMaximumSize(scrollArea.getPreferredSize());
        scrollArea.setBounds(15,20+height+space,650,1000);
        area.setFont(new Font("幼圆",Font.BOLD,15));

        scrollArea.setHorizontalScrollBarPolicy(
                JScrollPane.HORIZONTAL_SCROLLBAR_AS_NEEDED);
        scrollArea.setVerticalScrollBarPolicy(
                JScrollPane.VERTICAL_SCROLLBAR_AS_NEEDED);
        area.setEditable(false);
        panel.add(scrollArea);
        //eg
        //分别设置水平和垂直滚动条自动出现
//        js.setHorizontalScrollBarPolicy(
//                JScrollPane.HORIZONTAL_SCROLLBAR_AS_NEEDED);
//        js.setVerticalScrollBarPolicy(
//                JScrollPane.VERTICAL_SCROLLBAR_AS_NEEDED);
//        分别设置水平和垂直滚动条总是出现
//        js.setHorizontalScrollBarPolicy(
//                JScrollPane.HORIZONTAL_SCROLLBAR_ALWAYS);
//        js.setVerticalScrollBarPolicy(
//                JScrollPane.VERTICAL_SCROLLBAR_ALWAYS);
//        分别设置水平和垂直滚动条总是隐藏
//        js.setHorizontalScrollBarPolicy(
//                JScrollPane.HORIZONTAL_SCROLLBAR_NEVER);
//        js.setVerticalScrollBarPolicy(
//                JScrollPane.VERTICAL_SCROLLBAR_NEVER)

        //area查询历史
        area.addKeyListener(new KeyAdapter() {
            @Override
            public void keyPressed(KeyEvent e) {
                if(e.getKeyCode() == KeyEvent.VK_DELETE){
                    System.out.println("清空监控界面");
                    area.setText("");
                }
            }
        });

        topicBox.addItemListener(new ItemListener() {
            @Override
            public void itemStateChanged(ItemEvent e) {
                topicText.setText("");//重置
                if(topicBox.getSelectedIndex()!=0){
                    String key = topicBox.getSelectedItem().toString().trim();
                    List<String> topics = topicMap.get(key);
                    StringBuilder sb = new StringBuilder();
                    for(int i = 0 ; i < topics.size() ;i++){
                        if(i!= 0)
                            sb.append(",");
                        sb.append(topics.get(i));
                    }
                    topicText.setText(sb.toString().trim());
                }
            }
        });

        stopButton.addActionListener(new ActionListener() {
            @Override
            public void actionPerformed(ActionEvent e) {
                //锁按钮
                if(isMonitor){
                    isMonitor = false;
                    stopButton.setVisible(false);
                    startButton.setVisible(true);
                    topicText.setEnabled(true);
                    topicBox.setEnabled(true);
                    if(flush != null){
                        flush.stop();
                    }
                    if(reciving != null){
                        reciving.stop();
                    }
                    statusText.setText("监听已停止");
                }
            }
        });

        startButton.addActionListener(new ActionListener() {
            @Override
            public void actionPerformed(ActionEvent e) {

                //开多主题监听设置,数据隔离
                String topicStr = topicText.getText();
                String[] topicArray = topicStr.split(",");
                //开启监听
                if("".equals(topicStr)){
                    System.out.println("主题为空");
                    JOptionPane.showMessageDialog(null,"主题为空","kafkaSend",WARNING_MESSAGE);
                    return;
                }

                //锁按钮
                if(!isMonitor){
                    isMonitor = true;
                    stopButton.setVisible(true);
                    startButton.setVisible(false);
                    topicText.setEnabled(false);
                    topicBox.setEnabled(false);

                    List<ConsumerThread> consumerThreads = recived.reciveConsumer(topicArray);
                    for(ConsumerThread c:consumerThreads){
                        reciving = new Thread(c);
                        reciving.start();
                    }

                    //读
                    flush = new Thread(new Runnable() {
                        @Override
                        public void run() {
                            RecivedDataManger<ConsumerRecord<String, String>> instance = RecivedDataManger.getInstance();
                            Integer i = 0;
                            Integer j = 0;
                            while (true) {
                                Queue<ConsumerRecord<String, String>> read = instance.getRead();
                                Queue<ConsumerRecord<String, String>> write = instance.getWrite();
                                while (read.isEmpty()) {
                                    //只有当读队列为空且写队列大小超过1000时才进行交换。这样做的好处一是可以避免交换频率过高，二是保证一次获取一定量的实例
                                    try {
                                        if (write.size() >= 1) {
                                            instance.swap();
                                        } else {
                                            String text = area.getText();
                                            if (!"".equals(text) && i >= 10000) {
                                                //重置监听和清除数据
                                                area.setText("");
                                                i = 0;
                                            }
                                            statusText.setText("监听" + i++ + "s");
                                            Thread.sleep(1000);
                                        }
                                    } catch (Exception e) {
                                        e.printStackTrace();
                                    }
                                    read = instance.getRead();
                                }
                                //批量消费
                                while (!read.isEmpty()) {
                                    ConsumerRecord<String, String> poll = read.poll();
                                    area.insert("=================================================================================================\n", 0);
                                    String head = poll.topic() +"|"+poll.partition()+ "|" + poll.offset() + "|" + poll.key() +"|"+ Instant.ofEpochMilli(poll.timestamp()).atZone(ZoneId.systemDefault())+ "\n";
                                    String body = FormatUtil.format(poll.value() + "\n");
                                    if(j >= 10000){
                                        area.setText("");//清空,防止内存一处
                                        j = 0;
                                    }
                                    area.insert(body, 0);//每次都写到第一位
                                    area.insert(head, 0);//每次都写到第一位
                                    j++;
                                    System.out.println("---------消费------：" + head);
                                    log.info(head);//add 日志记录
                                    //1.将消息存到excel
                                    //ExcelService.saveExcel(poll);
                                    //2.将消息写到text中
                                    FileService.saveFile(head,poll);
                                }
                            }
                        }
                    });
                    flush.start();
                }


            }
        });

    }

}
